package com.nba;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 任务1（1）数据清洗要求
 * NBA的历史较为久远，从1947年至2019年的这段时间里，一些球队已经不存在了（例如：芝加哥牡鹿队），
 * 还有部分球队的队名发生了变化（例如：明尼阿波利斯湖人队，现在的名称是洛杉矶湖人队）；
 * 所以，对于已经不存在的球队，继续保存其名称，不做修改；
 * 但是已经更改名称的球队，需要映射为现在球队的名称；
 * 另外，因为要对球队进行东西分区的统计，所以要对球队添加东西分区的标识。
 */
public class DataClear {

    //建立任务(作业)，设置作业，提交作业
    public static void main(String[] args) throws Exception {

        //第一步：建立job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);//建立作业
        //第二步：设置驱动类
        job.setJarByClass(DataClear.class);  //绑定作业的类
        job.setJobName("DataClearNBA"); //设置作业名称
        //第三步：设置Mapper类
        job.setMapperClass(DataClear.MyMapper.class); //设置Mapper类
        job.setMapOutputKeyClass(NullWritable.class); //设置Map的输出的key的类型
        job.setMapOutputValueClass(Text.class); //设置Map的输出的value类型
        job.setNumReduceTasks(0);//不用设置Reduce,则无Reduce环节

        //第五步:设置输入和输出文件的路径
        // hadoop jar ....jar ..类名 分析的源文件的路径  输出的结果文件的路径
        FileInputFormat.setInputPaths(job, new Path("file/NBA/dataclear/nba.csv"));  //设置要分析的源文件路径
        FileOutputFormat.setOutputPath(job, new Path("file/NBA/dataclear/out")); //设置要输出的结果文件路径

//        //进行打包用
//        FileInputFormat.setInputPaths(job,new Path(args[0]));  //设置要分析的源文件路径
//        FileOutputFormat.setOutputPath(job,new Path(args[1])); //设置要输出的结果文件路径
        //第六步：提交程序
        job.waitForCompletion(true);
    }

    /**
     * NullWritable, Text :输出的结果：1947,4.16-4.22,费城勇士队,4-1,芝加哥牡鹿队
     */
    public static class MyMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
        //新旧球队名称替换的Map，key:旧名字 value 新名字
        Map<String, String> nameMap = new HashMap<String, String>();
        //东西分区球队的映射
        Map<String, List<String>> areaMap = new HashMap<String, List<String>>();

        Text newValue = new Text();//记录map输出的value值的

        /**
         * 在map方法执行之前，做预处理，将新旧球队名称的替换Map，哪些球队是东区的，哪些是西区的映射
         *
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void setup(Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
            //新旧球队名单，key:旧的球队名 value:新球队名
            nameMap.put("费城勇士队", "金州勇士队");
            nameMap.put("旧金山勇士队", "金州勇士队");
            nameMap.put("明尼阿波利斯湖人队", "洛杉矶湖人队");
            nameMap.put("塞拉库斯民族队", "费城76人队");
            nameMap.put("罗切斯特皇家队", "萨克拉门托国王队");
            nameMap.put("圣路易斯老鹰队", "亚特兰大老鹰队");
            nameMap.put("华盛顿子弹队", "华盛顿奇才队");
            nameMap.put("巴尔的摩子弹队", "华盛顿子弹队");
            nameMap.put("西雅图超音速队", "俄克拉荷马城雷霆队");
            nameMap.put("福特维恩活塞队", "底特律活塞队");
            nameMap.put("新泽西网队", "布鲁克林篮网队");
            nameMap.put("达拉斯小牛队", "达拉斯独行侠队");
            //东城区的球队名单
            List<String> eastList = new ArrayList<String>();
            eastList.add("亚特兰大老鹰队");
            eastList.add("夏洛特黄蜂队");
            eastList.add("迈阿密热火队");
            eastList.add("奥兰多魔术队");
            eastList.add("华盛顿奇才队");
            eastList.add("波士顿凯尔特人队");
            eastList.add("布鲁克林篮网队");
            eastList.add("纽约尼克斯队");
            eastList.add("费城76人队");
            eastList.add("多伦多猛龙队");
            eastList.add("芝加哥公牛队");
            eastList.add("克里夫兰骑士队");
            eastList.add("底特律活塞队");
            eastList.add("印第安纳步行者队");
            eastList.add("密尔沃基雄鹿队");
            //西城区的球队名单
            List<String> westList = new ArrayList<String>();
            westList.add("达拉斯独行侠队");
            westList.add("休斯顿火箭队");
            westList.add("孟菲斯灰熊队");
            westList.add("圣安东尼奥马刺队");
            westList.add("丹佛掘金队");
            westList.add("明尼苏拉森林狼队");
            westList.add("俄克拉荷马城雷霆队");
            westList.add("波特兰开拓者队");
            westList.add("犹他爵士队");
            westList.add("金州勇士队");
            westList.add("洛杉矶湖人队");
            westList.add("菲尼克斯太阳队");

            areaMap.put("east", eastList);
            areaMap.put("west", westList);
        }

        /**
         * 数据清洗
         *
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
            //使用,进行拆分字段，例如：   1947,4.16-4.22,费城勇士队,4-1,芝加哥牡鹿队,
            String[] split = value.toString().split(",");
            //(1)获取冠军队的名字
            String champPin = split[2];
            //获取亚军队的名称
            String second = split[4];
            //进行新旧球队名称的替换
            String newName = nameMap.get(champPin);
            split[2] = newName == null ? champPin : newName;
            String newName2 = nameMap.get(second);
            split[4] = newName2 == null ? second : newName2;

            //(2)1970年以后的球队，添加东西分区
            int year = Integer.parseInt(split[0]); //取出年份
            String areaFlag = "";//东西队的标志
            //1970之后的球队添加W E标志
            if (year >= 1970) {
                List<String> eastList = areaMap.get("east");//东区队的列表
                List<String> westList = areaMap.get("west");//西区队的列表
                if (eastList.contains(champPin)) {
                    areaFlag = "E";
                } else if (westList.contains(champPin)) {
                    areaFlag = "W";
                }
                //样例值：1974,4.28-5.12,波士顿凯尔特人队,4-3,密尔沃基雄鹿队,约翰·哈夫利切克,E
                newValue.set(strArrToString(split) + "," + areaFlag); //将String转为Text
            } else {
                newValue.set(strArrToString(split)); //将String转为Text
            }
            context.write(NullWritable.get(), newValue);
        }

        //字符串拼接，样例：1974,4.28-5.12,波士顿凯尔特人队,4-3,密尔沃基雄鹿队,约翰·哈夫利切克
        private String strArrToString(String[] split) {
            StringBuilder builder = new StringBuilder();
            for (int i = 0; i < split.length; i++) {
                builder.append(split[i]);
                if (i != split.length - 1) {
                    builder.append(",");
                }
            }
            return builder.toString();
        }
    }
}